package com.ai.bitcoin.queue;

import com.ai.bitcoin.exception.ThreadShutDownException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class MessageQueue {

    private static Logger logger = LoggerFactory
            .getLogger(MessageQueue.class);

    private BlockingQueue<String> queue;

    private boolean init = false;

    public MessageQueue(int capacity, boolean ret) {
        this.initial(capacity,ret);
    }

    public void initial(int capacity, boolean ret) {
        queue = new LinkedBlockingQueue<String>(capacity);
        init = ret;
    }

    public void send(String data) {
        if(init && data != null) {
//            String meta = JacksonUtils.toJson(data);
            //add 放不进去抛异常     offer 放不进去返回false  put 放不进去线程阻塞
            if(!queue.offer(data)) {
                //做补偿
                logger.error(data);
            }
        }
    }

    public String receive() {
        try {
            //poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止
            return queue.take();
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
            throw new ThreadShutDownException("message thread 线程终止");
        }
    }

}